/**
 * Copyright 2009 Erick Armbrust 
 * Licensed under the Apache License, Version 2.0 (the "License"); 
 * you may not use this file except in compliance with the License. 
 * You may obtain a copy of the License at 
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0 
 *   
 * Unless required by applicable law or agreed to in writing, software 
 * distributed under the License is distributed on an "AS IS" BASIS, 
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 * See the License for the specific language governing permissions and 
 * limitations under the License. 
 */

package tapioca.dt;

import java.util.List;
import java.util.Map;

import com.google.appengine.api.datastore.Blob;
import com.google.appengine.api.datastore.DatastoreService;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.EntityNotFoundException;
import com.google.appengine.api.datastore.EntityTranslator;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.KeyFactory;
import com.google.appengine.api.datastore.Transaction;
import com.google.appengine.repackaged.com.google.common.base.Preconditions;
import com.google.appengine.repackaged.com.google.common.collect.Lists;
import com.google.storage.onestore.v3.OnestoreEntity.EntityProto;

/**
 * A wrapper around an App Engine datastore {@link Entity} that defines a structure for
 * distributed transaction shadow copies.  Shadow copies store put or delete actions to be
 * performed on the datastore when a distributed transaction is rolled forward.
 * 
 * @author armbrust@gmail.com (Erick Armbrust)
 */
public class ShadowEntity {

  static final String KIND = "DT__shadow_copy";
  static final String CREATION_TIME_PROP = "creationTime";
  static final String DT_KEY_PROP = "dt";
  static final String ACTION_TYPE_PROP = "actionType";
  static final String SHADOW_BLOB_PROP = "shadow";
  static final String DELETE_KEY_PROP = "deleteKey";
  
  private final Entity entity;
  private final Entity shadowed;
  private final Key dtKey;
  private final Key deleteKey;
  private final ActionType actionType;
  
  /**
   * Private constructor.  The provided static DAO functions should be used to create shadow
   * entities.
   * 
   * @param entity the {@link Entity} that stores this shadow entity.
   * @param shadowed the {@link Entity} that is shadowed.
   * @param dtKey the {@link Key} for the distributed transaction that "owns" this shadow. 
   * @param deleteKey the {@link Key} for the entity to be deleted.
   * @param actionType the type of action being shadowed.
   */
  private ShadowEntity(
      Entity entity, 
      Entity shadowed, 
      Key dtKey,
      Key deleteKey, 
      ActionType actionType) {
    this.entity = entity;
    this.shadowed = shadowed;
    this.dtKey = dtKey;
    this.deleteKey = deleteKey;
    this.actionType = actionType;
  }
  
  ///// Getters /////
  
  public Entity getEntity() {
    return entity;
  }

  public Entity getShadowed() {
    return shadowed;
  }

  public Key getKey() {
    return entity.getKey();
  }
  
  public Key getDtKey() {
    return dtKey;
  }

  public Key getDeleteKey() {
    return deleteKey;
  }

  public ActionType getActionType() {
    return actionType;
  }  
  
  ///// Static DAO functions /////
  
  /**
   * Creates a key for a shadow object.  Shadow object keys are unique to a DT/entity combination.
   * This allows the writes of shadow objects to be idempotent.
   * 
   * @param dtKey the {@link Key} for the distributed transaction that "owns" this shadow.
   * @param shadowedKey the {@link Key} whose create/update/delete is being shadowed.
   * @return the {@link Key} for the shadow object.
   */
  public static Key createShadowKey(Key dtKey, Key shadowedKey) {
    Preconditions.checkArgument(shadowedKey.isComplete(), "Shadowed entity key must be complete");
    return KeyFactory.createKey(shadowedKey, KIND, 
        KeyFactory.keyToString(dtKey) + "-" + KeyFactory.keyToString(shadowedKey));
  }
  
  /**
   * Creates a new shadow entity for a delete operation and saves it to the datastore.
   * 
   * @param ds the {@link DatastoreService} to use for this operation. 
   * @param dtKey the {@link Key} for the distributed transaction that "owns" this shadow.
   * @param shadowed the {@link Entity} whose create/update is being shadowed.
   * @return the created shadow entity.
   */
  public static ShadowEntity createAndSavePutShadow(
      DatastoreService ds, 
      Key dtKey, 
      Entity shadowed) {
    Preconditions.checkArgument(shadowed.getKey().isComplete(), 
        "Shadowed entity key must be complete");    
    Entity entity = new Entity(createShadowKey(dtKey, shadowed.getKey()));
    
    // Fill the shadow copy.
    EntityProto proto = EntityTranslator.convertToPb(shadowed);
    Blob blob = new Blob(proto.toByteArray());
    
    entity.setProperty(DT_KEY_PROP, dtKey);
    entity.setProperty(ACTION_TYPE_PROP, ActionType.PUT.toString());
    entity.setProperty(SHADOW_BLOB_PROP, blob);
    ds.put(entity);
    
    return new ShadowEntity(entity, shadowed, dtKey, null, ActionType.PUT);
  }
  
  /**
   * Creates a new shadow entity for a delete operation and saves it to the datastore.
   * 
   * @param ds the {@link DatastoreService} to use for this operation. 
   * @param dtKey the {@link Key for the distributed transaction that "owns" this shadow.
   * @param deleteKey the {@link Key} for the entity to be deleted.
   * @return the created shadow entity.
   */
  public static ShadowEntity createAndSaveDeleteShadow(
      DatastoreService ds, 
      Key dtKey, 
      Key deleteKey) {
    Preconditions.checkArgument(deleteKey.isComplete(), "Cannot delete an incomplete key.");
    
    Entity entity = new Entity(createShadowKey(dtKey, deleteKey));
    entity.setProperty(DT_KEY_PROP, dtKey);
    entity.setProperty(ACTION_TYPE_PROP, ActionType.DELETE.toString());
    entity.setProperty(DELETE_KEY_PROP, deleteKey);
    ds.put(entity);
    
    return new ShadowEntity(entity, null, dtKey, deleteKey, ActionType.DELETE);
  }
  
  private static ShadowEntity loadShadow(Entity entity) {
    // Extract all the fields from the entity.
    Key dtKey = (Key) entity.getProperty(DT_KEY_PROP);
    ActionType actionType = ActionType.valueOf(
        (String) entity.getProperty(ACTION_TYPE_PROP));
    Blob blob = (Blob) entity.getProperty(SHADOW_BLOB_PROP);
    Key deleteKey = (Key) entity.getProperty(DELETE_KEY_PROP);
    
    Entity shadowed = null;
    if (blob != null) {
      // Unpack the stored entity proto.
      EntityProto proto = new EntityProto();
      proto.parseFrom(blob.getBytes());
      shadowed = EntityTranslator.createFromPb(proto);
    }
    
    return new ShadowEntity(entity, shadowed, dtKey, deleteKey, actionType);    
  }
  
  /**
   * Loads a shadow entity from the datastore.
   * 
   * @param ds the {@link DatastoreService} to use for this operation.
   * @param txn the {@link Transaction} to use for this operation.
   * @param key the {@link Key} for the shadow entity.
   * @return the loaded shadow entity.
   * @throws EntityNotFoundException if the shadow entity does not exist in the datastore.
   */
  public static ShadowEntity loadShadow(DatastoreService ds, Transaction txn, Key key)
      throws EntityNotFoundException {
    return loadShadow(ds.get(txn, key));
  }
  
  public static List<ShadowEntity> loadShadows(
      DatastoreService ds, 
      Transaction txn, 
      List<Key> keys) {
    List<ShadowEntity> shadows = Lists.newArrayList();
    Map<Key, Entity> shadowEntities = ds.get(txn, keys);

    for (Entity entity : shadowEntities.values()) {
      shadows.add(loadShadow(entity));
    }
    
    return shadows;
  }
  
}
